package com.yet.container.registry.zookeeper;

import java.text.MessageFormat;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.stream.Collectors;

import org.apache.curator.RetryPolicy;
import org.apache.curator.ensemble.EnsembleProvider;
import org.apache.curator.framework.AuthInfo;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.api.CompressionProvider;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.curator.utils.ZookeeperFactory;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.Ordered;
import org.springframework.util.Base64Utils;
import org.springframework.util.StringUtils;

import com.yet.container.registry.RegistryCenter;
import com.yet.container.registry.RegistryMeta;
import com.yet.container.registry.RegistryMeta.Address;
import com.yet.container.registry.RegistryMeta.ServiceMeta;
import com.yet.exception.RegistryNotFoundException;
import com.yet.exception.ServiceException;
import com.yet.listener.RegistryEvent;
import com.yet.listener.RegistryEvent.Type;
import com.yet.serialization.HessianSerialization;
import com.yet.serialization.Serialization;

/**
 * zookeeper注册中心
 */
public class ZookeeperRegistryCenter implements RegistryCenter {

	public static final Logger logger = LoggerFactory.getLogger(ZookeeperRegistryCenter.class);

	private final String rootPath; // 根节点

	private CuratorProperties curatorProperties;// 配置参数

	private BeanFactory beanFactory;

	private CuratorFramework curatorFramework;

	private final String name;

	private Serialization serialization = new HessianSerialization();

	private volatile State state = new State("Zookeeper注册中心", true);

	private final ExecutorService executorService = Executors.newFixedThreadPool(3);

	private TreeCache rootPathCache;

	private ApplicationEventPublisher publisher;

	private static final MessageFormat MESSAGE_FORMAT = new MessageFormat("[name:{0}, rootPath:{1}, address:{2}]");

	private TreeCacheListener listener = (client, event) -> {
		Type type = RegistryEvent.Type.str2Type(event.getType());
		if (type != null) {
			ServiceMeta serviceMeta = getServiceMetaFromEvent(event);
			if (serviceMeta != null) {
				logger.debug("Zookeeper监听事件 :{},{}", event.getType().name(), event.getData().getPath());
				getPublisher().publishEvent(new RegistryEvent(getName(), serviceMeta, type));
			}
		}
	};

	public ZookeeperRegistryCenter(String rootPath, String name) {
		validateRootPath(rootPath);
		this.name = name;
		this.rootPath = rootPath;

	}

	public ZookeeperRegistryCenter(String rootPath, String name, CuratorFramework curatorFramework) {
		super();
		validateRootPath(rootPath);
		this.rootPath = rootPath;
		this.name = name;
		this.curatorFramework = curatorFramework;
	}

	@Override
	public void start() {
		if (curatorFramework == null) {
			curatorFramework = curatorFramework();
		}
		curatorFramework.start();
		try {
			createRootPath();
			this.rootPathCache = new TreeCache(curatorFramework, rootPath);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}

	}

	@Override
	public void stop() {
		CloseableUtils.closeQuietly(rootPathCache);
		CloseableUtils.closeQuietly(curatorFramework);
		state.setStarted(false);
	}

	@Override
	public void register(RegistryMeta meta) {
		ServiceMeta serviceMeta = meta.getServiceMeta();
		String path = path(rootPath, serviceMeta.getName(), serviceMeta.getVersion());
		try {
			Stat stat = curatorFramework.checkExists().creatingParentsIfNeeded().forPath(path);
			if (stat == null) {
				curatorFramework.create().forPath(path,
						serialization.serialize(Collections.singletonList(meta.getAddress())));
			} else {
				List<Address> addresses = findByPathWithNoCheck(path);
				if (!addresses.contains(meta.getAddress())) {
					addresses.add(meta.getAddress());
					curatorFramework.setData().forPath(path, serialization.serialize(addresses));
				}
			}
		} catch (Exception e) {
			throw new ServiceException(e);
		}
	}

	@Override
	public void unregister(RegistryMeta meta) {
		ServiceMeta serviceMeta = meta.getServiceMeta();
		Address address = meta.getAddress();
		String path = path(rootPath, serviceMeta.getName(), serviceMeta.getVersion());
		try {
			List<String> addressPathList = curatorFramework.getChildren().forPath(path);
			for (String addressPath : addressPathList) {
				if (address.equals(serialization
						.deserialize(curatorFramework.getData().forPath(path(path, addressPath)), Address.class))) {
					curatorFramework.delete().forPath(path(path, addressPath));
				}
			}
		} catch (Exception e) {
			throw new ServiceException(String.format("下架服务失败%s", serviceMeta), e);
		}
	}

	@Override
	public void subscribe(ServiceMeta serviceMeta) {
		rootPathCache.getListenable().addListener(listener, executorService);
		try {
			rootPathCache.start();
		} catch (Exception e) {
			throw new ServiceException(e);
		}
	}

	@Override
	public void unsubscribe(ServiceMeta serviceMeta) {
		rootPathCache.getListenable().removeListener(listener);
	}

	@Override
	public List<Address> lookup(ServiceMeta serviceMeta) {
		String path = path(rootPath, serviceMeta.getName(), serviceMeta.getVersion());
		try {
			Stat stat = curatorFramework.checkExists().forPath(path);
			if (stat != null) {
				return findByPathWithNoCheck(path);
			}
			throw new RegistryNotFoundException();
		} catch (Exception e) {
			throw new RuntimeException(String.format("找不到服务%s", serviceMeta), e);
		}
	}

	@Override
	public Map<ServiceMeta, List<Address>> load() {
		List<String> forPath;
		try {
			forPath = curatorFramework.getChildren().forPath(rootPath);
			return forPath.stream().flatMap(service -> {
				try {
					return curatorFramework.getChildren().forPath(path(rootPath, service)).stream()
							.map(path -> new ServiceMeta(service, path));
				} catch (Exception e) {
					return null;
				}
			}).filter(s -> s != null).collect(Collectors.toMap(s -> s, this::lookup));
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	@SuppressWarnings("unchecked")
	private List<Address> findByPathWithNoCheck(String path) throws Exception {
		return serialization.deserialize(curatorFramework.getData().forPath(path), List.class);
	}

	private static String path(String... path) {
		return String.join("/", path);
	}

	private void createRootPath() throws Exception {
		if (curatorFramework.checkExists().forPath(rootPath) == null) {
			curatorFramework.create().forPath(rootPath);
		}
	}

	@Override
	public String getName() {
		return name;
	}

	public void setCuratorProperties(CuratorProperties curatorProperties) {
		this.curatorProperties = curatorProperties;
	}

	public CuratorProperties getCuratorProperties() {
		return curatorProperties;
	}

	public void setBeanFactory(BeanFactory beanFactory) {
		this.beanFactory = beanFactory;
	}

	private CuratorFramework curatorFramework() {
		final CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
		if (StringUtils.hasText(curatorProperties.getConnectString())) {
			builder.connectString(curatorProperties.getConnectString());
		} else if (StringUtils.hasLength(curatorProperties.getEnsembleProviderRef())) {
			builder.ensembleProvider(
					beanFactory.getBean(curatorProperties.getEnsembleProviderRef(), EnsembleProvider.class));
		} else {
			throw new IllegalArgumentException("缺少连接信息[spring.curator.connection-string]");
		}

		if (StringUtils.hasLength(curatorProperties.getAclProviderRef())) {
			builder.aclProvider(beanFactory.getBean(curatorProperties.getAclProviderRef(), ACLProvider.class));
		}

		if (StringUtils.hasText(curatorProperties.getAuthInfosRef())) {
			@SuppressWarnings("unchecked")
			List<AuthInfo> authInfos = beanFactory.getBean(curatorProperties.getAuthInfosRef(), List.class);
			builder.authorization(authInfos);
		} else if (StringUtils.hasText(curatorProperties.getScheme())
				&& StringUtils.hasText(curatorProperties.getAuthBase64Str())) {
			builder.authorization(curatorProperties.getScheme(),
					Base64Utils.decodeFromString(curatorProperties.getAuthBase64Str()));
		}

		if (curatorProperties.getCanBeReadOnly() != null) {
			builder.canBeReadOnly(curatorProperties.getCanBeReadOnly());
		}

		if (curatorProperties.getUseContainerParentsIfAvailable() != null
				&& !curatorProperties.getUseContainerParentsIfAvailable()) {
			builder.dontUseContainerParents();
		}

		if (StringUtils.hasLength(curatorProperties.getCompressionProviderRef())) {
			builder.compressionProvider(
					beanFactory.getBean(curatorProperties.getCompressionProviderRef(), CompressionProvider.class));
		}

		if (curatorProperties.getDefaultDataBase64Str() != null) {
			builder.defaultData(Base64Utils.decodeFromString(curatorProperties.getDefaultDataBase64Str()));
		}

		if (StringUtils.hasText(curatorProperties.getNamespace())) {
			builder.namespace(curatorProperties.getNamespace());
		}
		RetryPolicy retryPolicy;
		try {
			retryPolicy = beanFactory.getBean(RetryPolicy.class);
		} catch (BeansException e) {
			retryPolicy = new ExponentialBackoffRetry(curatorProperties.getBaseSleepTimeMs(),
					curatorProperties.getMaxRetries());
		}
		builder.retryPolicy(retryPolicy);

		if (null != curatorProperties.getSessionTimeOutMs()) {
			builder.sessionTimeoutMs(curatorProperties.getSessionTimeOutMs());
		}

		if (null != curatorProperties.getConnectionTimeoutMs()) {
			builder.connectionTimeoutMs(curatorProperties.getConnectionTimeoutMs());
		}

		if (null != curatorProperties.getMaxCloseWaitMs()) {
			builder.maxCloseWaitMs(curatorProperties.getMaxCloseWaitMs());
		}

		if (StringUtils.hasLength(curatorProperties.getThreadFactoryRef())) {
			builder.threadFactory(beanFactory.getBean(curatorProperties.getThreadFactoryRef(), ThreadFactory.class));
		}

		if (StringUtils.hasLength(curatorProperties.getZookeeperFactoryRef())) {
			builder.zookeeperFactory(
					beanFactory.getBean(curatorProperties.getZookeeperFactoryRef(), ZookeeperFactory.class));
		}

		logger.info("Start curatorFramework -> {}, sessionTimeOutMs={}, connectionTimeoutMs={}",
				curatorProperties.getConnectString(), curatorProperties.getSessionTimeOutMs(),
				curatorProperties.getConnectionTimeoutMs());

		return builder.build();
	}

	public static void validateRootPath(String rootPath) {
		boolean vaild = org.apache.commons.lang3.StringUtils.isNotBlank(rootPath) && rootPath.startsWith("/")
				&& !rootPath.endsWith("/");
		if (!vaild) {
			throw new IllegalArgumentException("错误的根节点:" + rootPath);
		}
	}

	private ServiceMeta getServiceMetaFromEvent(TreeCacheEvent event) {
		ChildData data = event.getData();
		String path = data.getPath();// rpc-root/name/version/
		String serviceStr = path.replaceFirst(rootPath, "");
		String[] parts = serviceStr.split("/");
		if (parts.length == 3) {
			ServiceMeta serviceMeta = new ServiceMeta(parts[1], parts[2]);
			return serviceMeta;
		}
		return null;
	}

	public Serialization getSerialization() {
		return serialization;
	}

	public void setSerialization(Serialization serialization) {
		this.serialization = serialization;
	}

	@Override
	public int getOrder() {
		return Ordered.HIGHEST_PRECEDENCE + 100;
	}

	public String getRootPath() {
		return rootPath;
	}

	@Override
	public State getState() {
		state.setInfo(MESSAGE_FORMAT
				.format(new Object[] { getName(), getRootPath(), getCuratorProperties().getConnectString() }));
		return state;
	}

	public ApplicationEventPublisher getPublisher() {
		return publisher;
	}

	public void setPublisher(ApplicationEventPublisher publisher) {
		this.publisher = publisher;
	}

}
